#pragma once

#include "../common/Message.hpp"
#include "../common/Net.hpp"
#include <unordered_set>

namespace rpc
{
    namespace server
    {
        class TopicManager
        {
        public:
            using  ptr = std::shared_ptr<TopicManager>;
            TopicManager(){}
            void onTopicRequest(const BaseConnection::ptr conn, const TopicRequest::ptr &req)
            {
                auto topic_operation = req->topicOption();
                bool ret = true;
                switch (topic_operation)
                {
                    case TOPIC_OP_TYPE::TOPIC_CREATE: // 主题创建
                        topicCreate(conn, req);
                        break;

                    case TOPIC_OP_TYPE::TOPIC_REMOVE: // 主题删除
                        topicRemove(conn, req);
                        break;

                    case TOPIC_OP_TYPE::TOPIC_SUBSCRIBE: // 主题订阅
                        ret = topicSubscribe(conn, req);
                        break;

                    case TOPIC_OP_TYPE::TOPIC_CANCEL: // 取消订阅
                        topicUnsubscribe(conn, req);
                        break;

                    case TOPIC_OP_TYPE::TOPIC_PUBLISH: // 主题消息推送
                        ret = topicMessagePush(conn, req);
                        break;

                    default:
                        response(conn, req, RCODE::RCODE_INVALID_OPERATION);
                        return;
                }

                if (ret == true) 
                    response(conn, req, RCODE::RCODE_OK);
                else 
                    // 订阅失败和消息推送失败，都是因为没有找到对应的主题
                    response(conn, req, RCODE::RCODE_NOT_FOUND_TOPIC);
            }

            void onConnShutdown(const BaseConnection::ptr &conn)
            {
                Subscriber::ptr user;
                std::vector<Topic::ptr> topics;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto user_it = _subscribers.find(conn);
                    if (user_it == _subscribers.end())
                    {
                        return;
                    }
                    user = user_it->second;

                    for (auto &topic_key : user->topics)
                    {
                        auto topic_it = _topics.find(topic_key);
                        if (topic_it == _topics.end())
                            continue;
                        topics.push_back(topic_it->second);
                    }

                    // 删除关联信息
                    _subscribers.erase(user_it);
                }
                for (auto &topic : topics)
                {
                    topic->removeSubscriber(user);
                }
            }
        private:
            // void TopicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg) // 函数名问题
            // void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            // {
            //     auto topic_rsp = MessageFactory::create<TopicResponse>();
            //     topic_rsp->setId(msg->id());
            //     topic_rsp->setMsgType(MSG_TYPE::RSP_TOPIC);
            //     topic_rsp->setRcode(RCODE::RCODE_OK);
            //     conn->send(topic_rsp);
            // }

            void response(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCODE rcode)
            {
                auto topic_rsp = MessageFactory::create<TopicResponse>();
                topic_rsp->setId(msg->id());
                topic_rsp->setMsgType(MSG_TYPE::RSP_TOPIC);
                topic_rsp->setRcode(rcode);
                conn->send(topic_rsp);
            }
            
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto topic_key = req->topicTheme();
                auto topic = std::make_shared<Topic>(topic_key);
                _topics[topic_key] = topic;
            }

            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                // 1. 判断当前主题是否存在：存在，取出订阅者；不存在，返回
                auto topic_key = req->topicTheme();
                
                std::unique_lock<std::mutex> lock(_mutex);
                auto topic_it = _topics.find(topic_key);
                if (topic_it == _topics.end())
                {
                    return;
                }

                auto subscribers = topic_it->second->subscribers;
                // 2. 订阅者取消订阅当前主题
                for (auto &user : subscribers)
                {
                    user->unsubscribe(topic_key);
                }

                // 3. 删除当前主题关联
                _topics.erase(topic_it);
            }

            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                auto topic_key = req->topicTheme();
                // 1. 主题存在则订阅，不存在则返回false
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_key);
                    if (topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;

                    // 2. 查找订阅者，存在则获取，不存在则创建
                    auto subscriber_it = _subscribers.find(conn);
                    if (subscriber_it == _subscribers.end())
                    {
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers[conn] = subscriber;
                    }
                    else 
                        subscriber = subscriber_it->second;
                }
                // 运行到这里，主题一定存在，且订阅者存在/被创建
                subscriber->subscribe(topic_key);
                topic->appendSubscriber(subscriber);

                // if (subscriber) subscriber->subscribe(topic_key);
                // if (subscriber && topic) topic->appendSubscriber(subscriber);
                return true;
            }

            void topicUnsubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                Subscriber::ptr user;
                Topic::ptr topic;
                auto topic_key = req->topicTheme();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto user_it = _subscribers.find(conn);
                    if (user_it != _subscribers.end())
                    {
                        user = user_it->second;
                    }

                    auto topic_it = _topics.find(topic_key);
                    if (topic_it != _topics.end())
                    {
                        topic = topic_it->second;
                    }
                }
                if (user)   user->unsubscribe(topic_key);
                if (user && topic) topic->removeSubscriber(user);
            }

            // 消息推送
            bool topicMessagePush(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                Topic::ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicTheme());
                    if (topic_it == _topics.end()) // 没有对应的主题
                    {
                        return false;
                    }
                    topic = topic_it->second;
                }

                topic->publishMessage(msg);
                return true;
            }

            // void topicUnsubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            // {
            //     // 查找订阅者，不存在则返回
            //     // 查找主题，不存在则返回
            //     Subscriber::ptr user;
            //     Topic::ptr topic;
            //     auto topic_key = req->topicTheme();
            //     {
            //         std::unique_lock<std::mutex> lock(_mutex);
            //         auto user_it = _subscribers.find(conn);
            //         if (user_it == _subscribers.end())
            //             return;
            //         user = user_it->second;

            //         auto topic_it = _topics.find(topic_key);
            //         if (topic_it == _topics.end())
            //             return;
            //         topic = topic_it->second;
            //         // user->unsubscribe(topic_key);
            //         // topic->removeSubscriber(user);
            //     }
            // }
        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> topics;

                Subscriber(const BaseConnection::ptr &c)
                    : conn(c)
                {}

                void subscribe(const std::string topic)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.insert(topic);
                }
                void unsubscribe(const std::string topic)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.erase(topic);
                }
            };

            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                std::string topic_name;
                std::unordered_set<Subscriber::ptr> subscribers;

                Topic(const std::string name)
                    : topic_name(name)
                {}

                void appendSubscriber(const Subscriber::ptr &user)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.insert(user);
                }
                void removeSubscriber(const Subscriber::ptr &user)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.erase(user);
                }
                void publishMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for (auto &user : subscribers)
                    {
                        user->conn->send(msg);
                    }
                }
            };
        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::ptr> _topics;
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;
        };
    }
}